package flink.connector.alarm;

import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.types.RowKind;

import static flink.connector.alarm.WechatAlarmConnectorOptions.*;

/**
 * {@link DynamicTableSink} via {@link WechatAlarmSink}
 *
 * @author jasonlee
 */
public class WechatAlarmDynamicTableSink implements DynamicTableSink {

    private final ReadableConfig options;

    public WechatAlarmDynamicTableSink(ReadableConfig options) {
        this.options = options;
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).build();
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        // 获取邮件的 URL,邮件类型
        final String host = options.getOptional(URL).get();
        final String msgType = options.getOptional(MSGTYPE).orElse("MARKDOWN");
        final String s = options.getOptional(MENTIONED_LIST).orElse(null);
        final String mentionedList = s == null ? MENTIONED_LIST.defaultValue() : s;
        WechatAlarmSink wechatAlarmSink = new WechatAlarmSink(host, msgType, mentionedList);
        return SinkFunctionProvider.of(wechatAlarmSink, 1);
    }

    @Override
    public DynamicTableSink copy() {
        return new WechatAlarmDynamicTableSink(this.options);
    }

    @Override
    public String asSummaryString() {
        return "wechat alarm table sink";
    }
}
